package com.wt.dwh.samples.flink;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;
import java.time.temporal.TemporalUnit;
import java.util.concurrent.TimeUnit;

public class WaterMarkExample {
    public static void main(String[] args) throws  Exception{
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStream<UserClickEvent> stream = env.fromCollection(UserClickEventSource.ELS).name("in-memory-input");
//        stream.assignTimestampsAndWatermarks(WatermarkStrategy.<UserClickEvent>forMonotonousTimestamps()
//                .withTimestampAssigner(new SerializableTimestampAssigner<UserClickEvent>() {
//                    @Override
//                    public long extractTimestamp(UserClickEvent event, long recordTimestamp) {
//                        return event.getTimestamp();
//                    }
//                })
//        );
        stream.assignTimestampsAndWatermarks(WatermarkStrategy.<UserClickEvent>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                .withTimestampAssigner((event,recordTimestamp)->event.getTimestamp())
        );

        stream.keyBy(value -> value.getUserName())
                .window(TumblingEventTimeWindows.of(Time.seconds(1)))
                // group by the tuple field "0" and sum up tuple field "1"
                .maxBy("timestamp")
                .print();
        env.execute();
    }
}
